﻿using System;
using System.Collections;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Reflection;
using System.Threading.Tasks;
using System.Threading;

namespace Tsual.DataPipeline
{
    public class DataPipeline<T, TMeta>
    {
        internal IPipeGenerator PipeGenerator { get; set; }
        private List<Type> Pipes { get; } = new List<Type>();
        private SemaphoreSlim slim = new SemaphoreSlim(1, 1);
        public DataPipeline(Func<Type, object> serviceResolveFunction)
        {
            PipeGenerator = new DefaultPipeGenerator(serviceResolveFunction);
        }

        public IEnumerable<Type> AllPipes => Pipes;

        public DataPipeline<T, TMeta> AddPipe<TPipe>()
            where TPipe : class, IDataPipeDescription<T, TMeta>
        {
            slim.Wait();
            try
            {
                if (typeof(TPipe).IsAbstract) throw new DataPipelineException("TPipe cannot be abstract");
                Pipes.Add(typeof(TPipe));
                return this;
            }
            finally
            {
                slim.Release();
            }

        }

        public DataPipeline<T, TMeta> InsertPipe<TPipe>()
            where TPipe : class, IDataPipeDescription<T, TMeta>
        {
            slim.Wait();
            try
            {
                if (typeof(TPipe).IsAbstract) throw new DataPipelineException("TPipe cannot be abstract");
                Pipes.Insert(0, typeof(TPipe));
                return this;
            }
            finally
            {
                slim.Release();
            }
        }

        public DataPipeline<T, TMeta> Concat(DataPipeline<T, TMeta> pipeline)
        {
            slim.Wait();
            pipeline.slim.Wait();
            try
            {
                pipeline.Pipes.ToList().ForEach(t => Pipes.Add(t));
                return this;
            }
            finally
            {
                slim.Release();
                pipeline.slim.Release();
            }
        }

        private DataPipelineExecutionResult<T, TMeta> finshExecution(IEnumerable<object> objs, DataPipelineExecutionResult<T, TMeta> result)
        {
            result.EndTime = DateTime.Now;
            if (objs != null)
                foreach (var obj in objs)
                    if (obj is IDisposable dis)
                    {
                        dis.Dispose();
                        result.CleanTypes.Add(dis.GetType());
                    }
            return result;
        }

        public async Task<DataPipelineExecutionResult<T, TMeta>> ReadAsync(TMeta meta, DataPipelineExecutionResult<T, TMeta> executionResult = null, PipelineReadPolicy pipelineReadPolicy = PipelineReadPolicy.CacheMode, bool ignoreReadError = true)
        {
            if (!Enum.IsDefined(typeof(PipelineReadPolicy), pipelineReadPolicy)) throw new DataPipelineException("Unkonw read policy");

            if (executionResult == null) executionResult = new DataPipelineExecutionResult<T, TMeta>(meta);

            var pipeIns = Pipes.Select(t => new KeyValuePair<Type, object>(t, null)).ToArray();

            //read
            executionResult.ReadStatus = DataPipelineExcutionStatus.Working;
            int pipeIndex;
            for (pipeIndex = 0; pipeIndex < pipeIns.Length; pipeIndex++)
            {
                var pipePair = pipeIns[pipeIndex];
                pipePair = new KeyValuePair<Type, object>(pipePair.Key, PipeGenerator.CreatePipeInstance(pipePair.Key));
                pipeIns[pipeIndex] = pipePair;

                {
                    var pipe = pipePair.Value as IDataPipeDescription<T, TMeta>;
                    try
                    {
                        if (!pipe.CanRead(meta))
                        {
                            executionResult.PipeResults.Enqueue(DataPipeReadResult<T>.Skip().SetPipeType(pipePair.Key));
                            continue;
                        }

                        var result = await pipe.Read(meta);
                        executionResult.PipeResults.Enqueue(result.SetPipeType(pipePair.Key));

                        if (result.Result == DataPipeReadResultEnum.Success)
                        {
                            executionResult.Data = result.Data;
                            executionResult.ReadStatus = DataPipelineExcutionStatus.Success;
                            break;
                        }
                        else if (!ignoreReadError && result.Result == DataPipeReadResultEnum.Error)
                        {
                            executionResult.ReadStatus = DataPipelineExcutionStatus.Error;
                            break;
                        }
                    }
                    catch (Exception ex)
                    {
                        executionResult.PipeResults.Enqueue(DataPipeReadResult<T>.Error(ex).SetPipeType(pipePair.Key));
                        if (!ignoreReadError)
                        {
                            executionResult.ReadStatus = DataPipelineExcutionStatus.Error;
                            break;
                        }

                    }
                }
            }

            //quit if read error
            if (executionResult.ReadStatus != DataPipelineExcutionStatus.Success)
            {
                if (executionResult.ReadStatus == DataPipelineExcutionStatus.Working)
                    executionResult.ReadStatus = DataPipelineExcutionStatus.NoResult;
                return finshExecution(pipeIns.Select(t => t.Value), executionResult);
            }

            //write
            executionResult.WriteStatus = DataPipelineExcutionStatus.Working;
            Queue<KeyValuePair<Type, object>> writeQueue = new Queue<KeyValuePair<Type, object>>();
            switch (pipelineReadPolicy)
            {
                case PipelineReadPolicy.CacheMode:
                    while (pipeIndex-- > 0) writeQueue.Enqueue(pipeIns[pipeIndex]);
                    break;
                case PipelineReadPolicy.PipeReadDepipeSet:
                    for (int i = 0; i < pipeIndex; i++) writeQueue.Enqueue(pipeIns[i]);
                    break;
                case PipelineReadPolicy.PipeReadAllSet:
                    foreach (var p in pipeIns) writeQueue.Enqueue(p);
                    break;
            }

            while (writeQueue.Count > 0)
            {
                var pipePair = writeQueue.Dequeue();

                if (pipePair.Value == null)
                {
                    pipePair = new KeyValuePair<Type, object>(pipePair.Key, PipeGenerator.CreatePipeInstance(pipePair.Key));
                }


                {
                    var pipe = pipePair.Value as IDataPipeDescription<T, TMeta>;
                    try
                    {
                        if (!pipe.CanWrite(executionResult.Data, meta))
                        {
                            executionResult.PipeResults.Enqueue(DataPipeWriteResult.Skip().SetPipeType(pipePair.Key));
                            continue;
                        }

                        var result = await pipe.Write(executionResult.Data, meta);
                        executionResult.PipeResults.Enqueue(result.SetPipeType(pipePair.Key));

                        if (result.Result == DataPipeWriteResultEnum.Error)
                        {
                            executionResult.WriteStatus = DataPipelineExcutionStatus.Error;
                            break;
                        }
                    }
                    catch (Exception ex)
                    {
                        executionResult.PipeResults.Enqueue(DataPipeWriteResult.Error(ex).SetPipeType(pipePair.Key));
                        executionResult.WriteStatus = DataPipelineExcutionStatus.Error;
                        break;
                    }
                }
            }

            //quit if write error
            if (executionResult.WriteStatus == DataPipelineExcutionStatus.Error)
            {
                return finshExecution(pipeIns.Select(t => t.Value), executionResult);
            }

            executionResult.WriteStatus = DataPipelineExcutionStatus.Success;
            return finshExecution(pipeIns.Select(t => t.Value), executionResult);
        }

        public async Task<DataPipelineExecutionResult<T, TMeta>> WriteAsync(T data, TMeta meta, DataPipelineExecutionResult<T, TMeta> executionResult = null, PipelineWritePolicy pipelineWritePolicy = PipelineWritePolicy.Forward)
        {
            if (!Enum.IsDefined(typeof(PipelineWritePolicy), pipelineWritePolicy)) throw new DataPipelineException("Unkonw write policy");

            if (executionResult == null) executionResult = new DataPipelineExecutionResult<T, TMeta>(meta);

            var pipeIns = Pipes.ToList();
            executionResult.Data = data;
            executionResult.WriteStatus = DataPipelineExcutionStatus.Working;
            if (pipelineWritePolicy == PipelineWritePolicy.Reverse)
                pipeIns = pipeIns.OrderByDescending(t => pipeIns.IndexOf(t)).ToList();

            var pipes = new List<IDataPipeDescription<T, TMeta>>();
            foreach (var pipeType in pipeIns)
            {
                var pipe = PipeGenerator.CreatePipeInstance(pipeType) as IDataPipeDescription<T, TMeta>;
                pipes.Add(pipe);
                try
                {
                    if (!pipe.CanWrite(executionResult.Data, meta))
                    {
                        executionResult.PipeResults.Enqueue(DataPipeWriteResult.Skip().SetPipeType(pipeType));
                        continue;
                    }

                    var result = await pipe.Write(data, meta);
                    executionResult.PipeResults.Enqueue(result.SetPipeType(pipeType));

                    if (result.Result == DataPipeWriteResultEnum.Error)
                    {
                        executionResult.WriteStatus = DataPipelineExcutionStatus.Error;
                        break;
                    }
                }
                catch (Exception ex)
                {
                    executionResult.PipeResults.Enqueue(DataPipeWriteResult.Error(ex).SetPipeType(pipeType));
                    executionResult.WriteStatus = DataPipelineExcutionStatus.Error;
                    break;
                }
            }

            //quit if write error
            if (executionResult.WriteStatus == DataPipelineExcutionStatus.Error)
            {
                return finshExecution(pipes, executionResult);
            }

            executionResult.WriteStatus = DataPipelineExcutionStatus.Success;
            return finshExecution(pipes, executionResult);
        }
    }

    public class SecondaryPipeline<T, TMeta, TPipeline, TSub, TMetaSub, TConvertor, TMetaConvertor> : IDataPipeDescription<T, TMeta>
        where TPipeline : DataPipeline<TSub, TMetaSub>
        where TMetaSub : IConvertible<PipelineReadPolicy>, IConvertible<PipelineWritePolicy>
        where TConvertor : class, IConvertor<T, TSub>, new()
        where TMetaConvertor : class, IConvertor<TMeta, TMetaSub>, new()
    {
        DataPipeline<TSub, TMetaSub> pipeline;
        IConvertor<T, TSub> convertor = new TConvertor();
        IConvertor<TMeta, TMetaSub> convertorMeta = new TMetaConvertor();

        internal SecondaryPipeline(IPipeGenerator pipeGenerator)
        {
            pipeline = pipeGenerator.CreatePipeInstance(typeof(TPipeline)) as DataPipeline<TSub, TMetaSub>;
        }

        public bool CanRead(TMeta meta) => true;

        public bool CanWrite(T data, TMeta meta) => true;

        public Task<DataPipeReadResult<T>> Read(TMeta pmeta) => Task.Run(async () =>
        {
            var meta = convertorMeta.Convert(pmeta);
            var policy = ((IConvertible<PipelineReadPolicy>)meta).Convert();
            var subresult = await pipeline.ReadAsync(meta, pipelineReadPolicy: policy);
            var data = subresult.GetResultOrThrowException();
            switch (subresult.ReadStatus)
            {
                case DataPipelineExcutionStatus.Success:
                    return DataPipeReadResult<T>.Success(convertor.Convert(subresult.Data));
                default:
                    return DataPipeReadResult<T>.Notfound();
            }
        });

        public Task<DataPipeWriteResult> Write(T data, TMeta pmeta) => Task.Run(() =>
        {
            var meta = convertorMeta.Convert(pmeta);
            var policy = ((IConvertible<PipelineWritePolicy>)meta).Convert();
            pipeline.WriteAsync(convertor.Convert(data), meta, pipelineWritePolicy: policy).GetResultOrThrowException();
            return DataPipeWriteResult.Success();
        });
    }

    public interface IDataPipeDescription<T, TMeta>
    {
        bool CanRead(TMeta meta);
        Task<DataPipeReadResult<T>> Read(TMeta meta);

        bool CanWrite(T data, TMeta meta);
        Task<DataPipeWriteResult> Write(T data, TMeta meta);
    }

    public class MetaConvertedPipe<T, TMeta, TMetaInner, TPipe, TConvertor> : IDataPipeDescription<T, TMeta>, IDisposable
        where TPipe : class, IDataPipeDescription<T, TMetaInner>
        where TConvertor : class, IHubConvertor<T, TMeta, TMetaInner>, new()
    {
        TPipe pipeInner;
        TConvertor metaConvertor = new TConvertor();
        public MetaConvertedPipe(IPipeGenerator pipeGenerator)
        {
            pipeInner = pipeGenerator.CreatePipeInstance(typeof(TPipe)) as TPipe;
        }

        public bool CanRead(TMeta meta)
        {
            return pipeInner.CanRead(metaConvertor.Convert(default, meta));
        }

        public bool CanWrite(T data, TMeta meta)
        {
            return pipeInner.CanWrite(data, metaConvertor.Convert(data, meta));
        }

        public void Dispose()
        {
            if (pipeInner is IDisposable dis) dis.Dispose();
        }

        public Task<DataPipeReadResult<T>> Read(TMeta meta)
        {
            return pipeInner.Read(metaConvertor.Convert(default, meta));
        }

        public Task<DataPipeWriteResult> Write(T data, TMeta meta)
        {
            return pipeInner.Write(data, metaConvertor.Convert(data, meta));
        }
    }

    /// <summary>
    /// Attach a pre-read pipe
    /// </summary>
    /// <typeparam name="T"></typeparam>
    /// <typeparam name="TInner"></typeparam>
    /// <typeparam name="TMeta"></typeparam>
    /// <typeparam name="TPipe0"></typeparam>
    /// <typeparam name="TPip1"></typeparam>
    public class CompositeReadPipe<T, TMeta, TInner, TMetaInner, TPipePreread, TPipeRead, TPipeWrite, TMetaConvertor> : IDataPipeDescription<T, TMeta>,IDisposable
        where TPipePreread : class, IDataPipeDescription<TInner, TMetaInner>
        where TPipeRead : class, IDataPipeDescription<T, TInner>
        where TPipeWrite : class, IDataPipeDescription<T, TMetaInner>
        where TMetaConvertor : class, IConvertor<TMeta, TMetaInner>, new()
    {
        IPipeGenerator generator { get; set; }

        public CompositeReadPipe(IPipeGenerator generator)
        {
            this.generator = generator;
            pipePreread = generator.CreatePipeInstance(typeof(TPipePreread)) as TPipePreread;
            pipeRead = generator.CreatePipeInstance(typeof(TPipeRead)) as TPipeRead;
            pipeWrite = generator.CreatePipeInstance(typeof(TPipeWrite)) as TPipeWrite;
        }

        TPipePreread pipePreread;
        TPipeRead pipeRead;
        TPipeWrite pipeWrite;

        TMetaConvertor metaConvertor = new TMetaConvertor();

        public bool CanRead(TMeta meta) => true;

        public bool CanWrite(T data, TMeta meta) => true;

        public async Task<DataPipeReadResult<T>> Read(TMeta plmeta)
        {
            var meta = metaConvertor.Convert(plmeta);
            if (!pipePreread.CanRead(meta))
                //throw new DataPipelineException($"{typeof(TPipePreread).FullName} must readable");
                return DataPipeReadResult<T>.Notfound();
            var prereadResult = await pipePreread.Read(meta);

            switch (prereadResult.Result)
            {
                case DataPipeReadResultEnum.Notfound:
                    return DataPipeReadResult<T>.Notfound();
                case DataPipeReadResultEnum.Error:
                    return DataPipeReadResult<T>.Notfound();
            }

            var pmeta = prereadResult.Data;
            if (!pipeRead.CanRead(pmeta))
                //throw new DataPipelineException($"{typeof(TPipeRead).FullName} must readable");
                return DataPipeReadResult<T>.Notfound();

            return await pipeRead.Read(pmeta);
        }

        public Task<DataPipeWriteResult> Write(T data, TMeta meta) => pipeWrite.Write(data, metaConvertor.Convert(meta));

        public void Dispose()
        {
            { if (pipePreread is IDisposable dis) dis.Dispose(); }
            { if (pipeRead is IDisposable dis) dis.Dispose(); }
            { if (pipeWrite is IDisposable dis) dis.Dispose(); }
        }
    }
    public interface IPipeGenerator
    {
        Func<Type, object> ServiceResolveFunction { get; }
        object CreatePipeInstance(ConstructorInfo ctor);
        object CreatePipeInstance(Type type);
    }

    internal class DefaultPipeGenerator : IPipeGenerator
    {
        public Func<Type, object> ServiceResolveFunction { get; set; }

        public DefaultPipeGenerator(Func<Type, object> serviceResolveFunction)
        {
            ServiceResolveFunction = serviceResolveFunction;
        }

        public object CreatePipeInstance(ConstructorInfo ctor)
        {
            return ctor.Invoke(ctor.GetParameters().Select(p => p.ParameterType == typeof(IPipeGenerator) ? this : ServiceResolveFunction(p.ParameterType)).ToArray());
        }

        public object CreatePipeInstance(Type type) => CreatePipeInstance(type.GetConstructors().Where(ctorT => ctorT.IsPublic).FirstOrDefault() ?? throw new TypeAccessException($"Type[{type.FullName}] has 0 public constructor"));
    }

    public class PipeObjectFactory : IReadOnlyDictionary<string, object>
    {
        private ConcurrentDictionary<string, object> map = new ConcurrentDictionary<string, object>();

        private PipeObjectFactory() { }

        public object this[string key] => ((IReadOnlyDictionary<string, object>)map)[key];

        public static PipeObjectFactory Instance { get; } = new PipeObjectFactory();

        public IEnumerable<string> Keys => ((IReadOnlyDictionary<string, object>)map).Keys;

        public IEnumerable<object> Values => ((IReadOnlyDictionary<string, object>)map).Values;

        public int Count => ((IReadOnlyDictionary<string, object>)map).Count;

        public bool ContainsKey(string key)
        {
            return ((IReadOnlyDictionary<string, object>)map).ContainsKey(key);
        }

        public IEnumerator<KeyValuePair<string, object>> GetEnumerator()
        {
            return ((IReadOnlyDictionary<string, object>)map).GetEnumerator();
        }

        public bool TryGetValue(string key, out object value)
        {
            return ((IReadOnlyDictionary<string, object>)map).TryGetValue(key, out value);
        }



        public object AddOrUpdate(string key, Func<string, object> addValueFactory, Func<string, object, object> updateValueFactory) => map.AddOrUpdate(key, addValueFactory, updateValueFactory);

        public object AddOrUpdate(string key, object addValue, Func<string, object, object> updateValueFactory) => map.AddOrUpdate(key, addValue, updateValueFactory);

        IEnumerator IEnumerable.GetEnumerator()
        {
            return ((IReadOnlyDictionary<string, object>)map).GetEnumerator();
        }
    }
}
